/**
 * Copyright 1996 self.com.au
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License. (http://www.apache.org/licenses/LICENSE-2.0)
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed 
 * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for 
 * the specific language governing permissions and limitations under the License.
 *
 * <dhamma-message src="Atisha (11th century Tibetan Buddhist master)">
 * 	The greatest achievement is selflessness. The greatest worth is self-mastery.
 * 	The greatest quality is seeking to serve others. The greatest precept is continual awareness.
 * 	The greatest medicine is the emptiness of everything. The greatest action is not conforming with the worlds ways.
 * 	The greatest magic is transmuting the passions. The greatest generosity is non-attachment.
 * 	The greatest goodness is a peaceful mind. The greatest patience is humility.
 * 	The greatest effort is not concerned with results. The greatest meditation is a mind that lets go.
 * 	The greatest wisdom is seeing through appearances. 
 * </dhamma-message>
 *
 * @author Ashin Wimalajeewa
 *
 * Note, this is a very old runtime library that I have evolved since 1996. I rarely use it
 * now because the Apache Commons project has made it obsolete.
 */
package self.oss.lucene.store.db;

/**
 * Copyright 2004 The Apache Software Foundation
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;

import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.InputStream;
import org.apache.lucene.store.Lock;
import org.apache.lucene.store.OutputStream;

import self.lang.UidUtils;

/**
 * A blob-based {@link Directory} implementation. Use this implementation in
 * conjunction with the aidindex, aidindex_buffer & aidindex_queuelock database
 * objects (see DDL below).
 *
 * Requirements:
 * -------------
 *  - Relational database capable of supporting a blob-like field type that can
 *    store 1024 bytes. Note, this data will be:
 *    1) Read via
 *      PreparedStatement.getBytes( colNum ) method
 *    2) Written via
 *      PreparedStatement.setBytes( colNum, dataIn1024ByteArray ) method
 *    ** This field type will vary per RDBMS (see MySQL & Oracle examples below)
 *  - Write access to the JVM "java.io.tmpdir" directory
 *  - A JDBC Connection with AUTO-COMMIT set to TRUE
 *
 * Forces:
 * -------
 *  - An RDBMS directory will never be as fast as a Lucene's default FileSystem
 *    directory
 *  - Lucene's default RAM directory may be too taxing on large index files
 *  - A globally accessible directory is needed for application that run in a
 *    cluster
 *  - A new lock scheme is needed such that IndexReaders, IndexWriters &
 *    IndexSearches can all be invoked concurrently without suffering from
 *    time-outs
 *  - I tried to implement semaphores for locks with a non-AUTO-COMMIT connection
 *    for index writing but I was getting deadlocks. Reverted back to an AUTO-
 *    COMMIT JDBC connection
 *  - Combining FileIO with DML insert calls on close, seems to execute
 *    considerably faster than a pure DML-based OutputStream. You can test this
 *    yourself by replacing the "new TemporaryFileOutputStream( name );" with
 *    "new BlobRowsOutputStream( name );"
 *
 * Strategy
 * --------
 *  1. store Lucene's index meta-data in the aidindex table. Each row represents
 *     a Lucene index file
 *  2. store Lucene's index buffer in the aidindex_buffer table. Each row
 *     represents a 1024 byte array element modeled on the RAMDirectory
 *  3. store Lucene's locks in the aidindex_queuelock table. This way locks
 *     are preserved across JVMs in a clustered environment. Furthermore, lock
 *     requests are queued rather than arbitarily timing-out based Lucene's
 *     WRITE/COMMIT time-out properties
 *  4. when writing out streams, write the stream out to a unique temp file
 *     first. Once the stream is closed, persist the stream to the RDBMS and
 *     finally delete the temp file
 *
 *
 * DDL:
 * ----
 * Eg. in MySQL:

    create table aidindex (
      directory_name varchar(96) not null,
      name varchar(96) not null,
      lastmodified int(32) not null,
      length int(32) not null,
      primary key(directory_name, name)
    );

    create table aidindex_buffer (
      directory_name varchar(96) not null,
      name varchar(96) not null,
      id int(32) not null,
      buffer blob,
      primary key(directory_name, name, id)
    );

    create table aidindex_queuelock (
      directory_name varchar(96) not null,
      name varchar(96) not null,
      caller_id varchar(36) not null,
      primary key(directory_name, name, caller_id)
    );


 * Eg. in Oracle:

  CREATE TABLE "AIDINDEX" ("DIRECTORY_NAME" VARCHAR2(96 byte)
    NOT NULL, "NAME" VARCHAR2(96 byte) NOT NULL, "LASTMODIFIED"
    NUMBER(32) NOT NULL, "LENGTH" NUMBER(32) NOT NULL,
    CONSTRAINT "AIDINDEX_PK11085815654527" PRIMARY
    KEY("DIRECTORY_NAME", "NAME")
    );

  CREATE TABLE "AIDINDEX_BUFFER" ("DIRECTORY_NAME"
    VARCHAR2(96 byte) NOT NULL, "NAME" VARCHAR2(96 byte) NOT NULL,
    "ID" NUMBER(32) NOT NULL, "BUFFER" RAW(1024),
    CONSTRAINT "SYS_C003082" PRIMARY KEY("DIRECTORY_NAME", "NAME",
    "ID")
    );


  CREATE TABLE "AIDINDEX_QUEUELOCK" ("DIRECTORY_NAME"
    VARCHAR2(96 byte) NOT NULL, "NAME" VARCHAR2(96 byte) NOT NULL,
    "CALLER_ID" VARCHAR2(36 byte) NOT NULL,
    CONSTRAINT "SYS_C003086_1" PRIMARY KEY("DIRECTORY_NAME",
    "NAME", "CALLER_ID")
    );
 */

public class BlobDirectory extends Directory {
  final static public int BUFFER_SIZE = 1024;

  static public interface Store {
    public String getIndexTableName();
    public String getIndexQueueLockTableName();
    public String getBufferTableName();
    public String getDirectoryName();
    public Connection getConnection();
    public void closeConnection( Connection con );
  }

  private Store store;

  public BlobDirectory( Store source ) {
    store = source;
  }

  public String[] list() throws IOException {
// System.out.println( Thread.currentThread().getName() + ") BlobDirectory.list");
    ArrayList names = new ArrayList();
    try {
      Connection conn = store.getConnection();
      try {
        PreparedStatement stmt = conn.prepareStatement(
            "select name from " + store.getIndexTableName() + " where directory_name=?");
        try {
          stmt.setString( 1, store.getDirectoryName() );
          ResultSet rs = stmt.executeQuery();
          try {
            while( rs.next() )
                names.add( rs.getString(1) );
          } finally {
            rs.close();
          }
        } finally {
          stmt.close();
        }
      } finally {
          store.closeConnection( conn );
      }
    } catch ( SQLException e ) {
      throw new IOException( e.getMessage() );
    }
    String[] ret = new String[ names.size() ];
    names.toArray( ret );
    return ret;
  }

  public boolean fileExists(String name) throws IOException {
// System.out.println(Thread.currentThread().getName() + ")BlobDirectory.fileExists - " + name );
    try {
      Connection conn = store.getConnection();
      try {
        PreparedStatement stmt = conn.prepareStatement(
            "select name from " + store.getIndexTableName() + " where name=? and directory_name=?");
        try {
          stmt.setString(1, name);
          stmt.setString(2, store.getDirectoryName());
          ResultSet rs = stmt.executeQuery();
          try {
            boolean ret = rs.next();
            return ret;
          } finally {
            rs.close();
          }
        } finally {
          stmt.close();
        }
      } finally {
        store.closeConnection( conn );
      }
    } catch ( SQLException err ) {
      throw new IOException( err.getMessage() );
    }
  }

  public long fileModified(String name) throws IOException {
// System.out.println( Thread.currentThread().getName() +  ")BlobDirectory.fileModified - " + name );
    try {
      Connection conn = store.getConnection();
      try {
        PreparedStatement stmt = conn.prepareStatement(
            "select lastmodified from " + store.getIndexTableName() + " where name=? and directory_name=?");
        try {
          stmt.setString(1, name);
          stmt.setString(2, store.getDirectoryName());
          ResultSet rs = stmt.executeQuery();
          try {
            if ( rs.next() )
              return rs.getTimestamp( 1 ).getTime();
            throw new IOException("File not found: " + name);
          } finally {
            rs.close();
          }
        } finally {
          stmt.close();
        }
      } finally {
        store.closeConnection( conn );
      }
    } catch ( SQLException err ) {
      throw new IOException( err.getMessage() );
    }
  }

  public void touchFile(String name) throws IOException {
// System.out.println( Thread.currentThread().getName() + ")BlobDirectory.touchFile - " + name );
    try {
      Connection conn = store.getConnection();
      try {
        PreparedStatement stmt = conn.prepareStatement(
            "update " + store.getIndexTableName() + " set lastmodified=? where name=? and directory_name=?");
        try {
          stmt.setTimestamp(1, new java.sql.Timestamp(System.currentTimeMillis()) );
          stmt.setString(2, name);
          stmt.setString(3, store.getDirectoryName());
          if ( stmt.executeUpdate() < 1 )
            throw new IOException( "File not found: " + name );
        } finally {
          stmt.close();
        }
      } finally {
        store.closeConnection( conn );
      }
    } catch ( SQLException err ) {
      throw new IOException( err.getMessage() );
    }
  }

  public long fileLength(String name) throws IOException {
// System.out.println(Thread.currentThread().getName() + ")BlobDirectory.fileLength - " + name );
    try {
      Connection conn = store.getConnection();
      try {
        PreparedStatement stmt = conn.prepareStatement(
            "select length from " + store.getIndexTableName() + " where name=? and directory_name=?");
        try {
          stmt.setString(1, name);
          stmt.setString(2, store.getDirectoryName());
          ResultSet rs = stmt.executeQuery();
          try {
            if( rs.next() )
              return rs.getLong( 1 );
            throw new IOException("File not found: " + name);
          } finally {
            rs.close();
          }
        } finally {
          stmt.close();
        }
      } finally {
        store.closeConnection( conn );
      }
    } catch ( SQLException err ) {
      throw new IOException( err.getMessage() );
    }
  }

  public void deleteFile(String name) throws IOException {
// System.out.println(Thread.currentThread().getName() + ")<BlobDirectory.deleteFile - " + name + ">" );
    try {
      Connection conn = store.getConnection();
      try {
        PreparedStatement stmt = conn.prepareStatement(
            "delete from " + store.getIndexTableName() + " where name=? and directory_name=?");
        try {
          stmt.setString(1, name);
          stmt.setString(2, store.getDirectoryName());
          if (stmt.executeUpdate() < 1 )
            throw new IOException("File not found: " + name);
        } finally {
          stmt.close();
        }
        stmt = conn.prepareStatement(
            "delete from " + store.getBufferTableName() + " where name=? and directory_name=?");
        try {
          stmt.setString(1, name);
          stmt.setString(2, store.getDirectoryName());
          stmt.executeUpdate();
        } finally {
          stmt.close();
        }
      } finally {
        store.closeConnection( conn );
      }
    } catch ( SQLException err ) {
      throw new IOException( err.getMessage() );
    }
  }

  public void renameFile(String from, String to) throws IOException {
// System.out.println(Thread.currentThread().getName() + ")BlobDirectory.renameFile - " + from + " -> " + to );
    try {
      Connection conn = store.getConnection();
      try {
        PreparedStatement stmt = conn.prepareStatement(
            "delete from " + store.getBufferTableName() + " where name=? and directory_name=?");
        try {
          stmt.setString(1, to);
          stmt.setString(2, store.getDirectoryName());
          stmt.executeUpdate();
        } finally {
          stmt.close();
        }
        stmt = conn.prepareStatement(
            "update " + store.getBufferTableName() + " set name=? where name=? and directory_name=?");
        try {
          stmt.setString(1, to);
          stmt.setString(2, from);
          stmt.setString(3, store.getDirectoryName());
          stmt.executeUpdate();
        } finally {
          stmt.close();
        }
        stmt = conn.prepareStatement(
            "delete from " + store.getIndexTableName() + " where name=? and directory_name=?");
        try {
          stmt.setString(1, to);
          stmt.setString(2, store.getDirectoryName());
          stmt.executeUpdate();
        } finally {
          stmt.close();
        }
        stmt = conn.prepareStatement(
            "update " + store.getIndexTableName() + " set name=? where name=? and directory_name=?");
        try {
          stmt.setString(1, to);
          stmt.setString(2, from);
          stmt.setString(3, store.getDirectoryName());
          if (stmt.executeUpdate() < 1 )
            throw new IOException("File not found: " + from );
        } finally {
          stmt.close();
        }
      } finally {
        store.closeConnection( conn );
      }
    } catch ( SQLException err ) {
      err.printStackTrace();
      throw new IOException( err.getMessage() );
    }
  }

  public OutputStream createFile(String name) throws IOException {
//System.out.println(Thread.currentThread().getName() + ")BlobDirectory.createFile - " + name );
    try {
      Connection conn = store.getConnection();
      try {
        PreparedStatement stmt = conn.prepareStatement(
            "insert into " + store.getIndexTableName() + " (name, lastmodified, length, directory_name) values (?,?,0,?)");
        try {
          stmt.setString(1, name);
          stmt.setTimestamp(2, new java.sql.Timestamp(System.currentTimeMillis()) );
          stmt.setString(3, store.getDirectoryName());
          stmt.executeUpdate();
        } finally {
          stmt.close();
        }
      } finally {
        store.closeConnection( conn );
      }
    } catch ( SQLException err ) {
      throw new IOException( err.getMessage() );
    }
    return new TemporaryFileOutputStream( name );
//    return new BlobRowsOutputStream( name );
  }

  public InputStream openFile(String name) throws IOException {
//System.out.println(Thread.currentThread().getName() + ")BlobDirectory.openFile - " + name );
    long len;
    try {
      Connection conn = store.getConnection();
      try {
        PreparedStatement stmt = conn.prepareStatement(
            "select length from " + store.getIndexTableName() + " where name=? and directory_name=?");
        try {
          stmt.setString(1, name);
          stmt.setString(2, store.getDirectoryName());
          ResultSet rs = stmt.executeQuery();
          try {
            if ( !rs.next() )
              throw new IOException( "File not found: " + name );
            len = rs.getLong( 1 );
          } finally {
            rs.close();
          }
        } finally {
          stmt.close();
        }
      } finally {
        store.closeConnection( conn );
      }
    } catch ( SQLException err ) {
      throw new IOException( err.getMessage() );
    }
    return new BlobRowsInputStream( name, len );
  }

  public Lock makeLock(final String name) {
//  	System.out.println( Thread.currentThread().getName() + ")BlobDirectory.makeLock name=" + name );
    return new QueueLock( name );
  }

  public void close() {
//System.out.print("(close,th=" +Thread.currentThread().hashCode() + ")");
  }

  private byte[] getBuffer( Connection conn, String name, int buffId ) throws SQLException, IOException {
//  	System.out.println( Thread.currentThread().getName() + ")BlobDirectory.getBuffer, name=" + name + ", buffId=" + buffId );
    PreparedStatement stmt = conn.prepareStatement( "select buffer from " + store.getBufferTableName() + " where id=? and name=? and directory_name=?" );
    try {
      stmt.setInt(1, buffId);
      stmt.setString(2, name);
      stmt.setString(3, store.getDirectoryName());
      ResultSet rs = stmt.executeQuery();
      try {
        if ( !rs.next() )
          throw new IOException( "buffer not found" );
        return rs.getBytes( 1 );
      } finally {
        rs.close();
      }
    } finally {
      stmt.close();
    }
  }

  private class QueueLock extends Lock {
    private String name;
    private String callerId;
    private boolean isWriteLock;

    public QueueLock( String filename ) {
      name = filename;
      isWriteLock = name.equals( IndexWriter.WRITE_LOCK_NAME );
    }

    /**
     * "write.lock" requests may be registered by the client for a reasonably
     * long duration. In such instances as two seperate threads invokings a
     * re-indexing function, the first that acquires the "write.lock" will cause
     * the second to block until the entire indexings package is complete.
     *
     * Hence, need a strategy that can keep the block alive. This strategy is
     * better realised by implementing a co-operative client lock use approach.
     * This is achieved by ensuring that the package size worked on by an
     * IndexWriter does not take an excessive amount of time (eg < 2 min). That
     * is, clients should ensure that the time that transpires between:
     * IndexWriter writer = new IndexWriter( indexStore, indexAnalyzer, false );
     *   ...
     * writer.close();
     * is less than 2 min.
     */
    public boolean obtain(long lockWaitTimeout) throws IOException {
//      int waitDuration = ( name.equals(IndexWriter.WRITE_LOCK_NAME)? 120: 60 );
/*
    Observations indicate that search's use only commit.locks. However, writing
    and deleting from the index requires both write.lock and commit.locks.
    Therefore, I have changed the waitDuration to FOREVER for a write.lock. This
    should force all pending updates to be flushed to the index, whilst still
    permitting searches.
*/

      int waitDuration = ( isWriteLock ? Integer.MAX_VALUE: 60 );
      float intervalDuration = LOCK_POLL_INTERVAL / 1000;
      waitDuration = (int) (waitDuration * intervalDuration);
      callerId = UidUtils.getNwgGuid();
      boolean locked = false;
      joinQueue();
//System.out.println( new java.util.Date() + "> " + callerId + " waitDuration = " + waitDuration + " [" + Thread.currentThread().getName() + "]:: " + name);
      double secCntr = 0;
      try {
        while ( secCntr < waitDuration ) {  // wait in queue
          if ( isFirstInQueue() ) {
            locked = true;
            break;
          }
          try { Thread.sleep(LOCK_POLL_INTERVAL); } catch (InterruptedException e) {
          	// nothing to do.
          }
          secCntr++;
        }
      } finally {
        if ( !locked ) {
          leaveQueue();
//System.out.println( new java.util.Date() + "> " + callerId + " TIMEDOUT and left queue [" + Thread.currentThread().getName() + "]:: " + name);
        }
      }
      return locked;
    }

    public boolean obtain() throws IOException {
      throw new UnsupportedOperationException();
    }

    public void release() {
      try {
        leaveQueue();
      } catch ( Exception err ) {
        throw new IllegalStateException( err.getMessage() );
      }
    }

    public boolean isLocked() {
      try {
        Connection conn = store.getConnection();
        try {
          PreparedStatement stmt = conn.prepareStatement(
              "select count(0) from " + store.getIndexQueueLockTableName() + " where directory_name=? and name=?");
          try {
            stmt.setString(1, store.getDirectoryName());
            stmt.setString(2, name);
            ResultSet rs = stmt.executeQuery();
            try {
              rs.next();
              int total = rs.getInt( 1 );
              return (total > 0);
            } finally {
              rs.close();
            }
          } finally {
            stmt.close();
          }
        } finally {
          store.closeConnection( conn );
        }
      } catch ( SQLException err ) {
        throw new IllegalStateException( err.getMessage() );
      }
    }

    private void joinQueue() throws IOException {
//System.out.println( Thread.currentThread().getName() + "::" + name + ")BlobDirectory::joinQueue (" + callerId + ")");
      try {
        Connection conn = store.getConnection();
        try {
          PreparedStatement stmt = conn.prepareStatement(
              "insert into " + store.getIndexQueueLockTableName() + " (directory_name, name, caller_id, arrival_time) values (?,?,?,?)");
//System.out.println( new java.util.Date() + "> " + callerId + " trying to join queue [" + Thread.currentThread().getName() + "]:: " + name);
          try {
            stmt.setString(1, store.getDirectoryName());
            stmt.setString(2, name);
            stmt.setString(3, callerId);
            stmt.setTimestamp( 4, new java.sql.Timestamp(System.currentTimeMillis()) );
            stmt.executeUpdate();
//System.out.println( new java.util.Date() + "> " + callerId + " has joined queue [" + Thread.currentThread().getName() + "]:: " + name);
          } finally {
            stmt.close();
          }
        } finally {
          store.closeConnection( conn );
        }
      } catch ( SQLException err ) {
//System.out.println( new java.util.Date() + "> " + callerId + " has had troubles joining the queue [" + Thread.currentThread().getName() + "]:: " + name);
        throw new IOException( err.getMessage() );
      }
    }

    private boolean isFirstInQueue() throws IOException {
      boolean leave = false;
      try {
        Connection conn = store.getConnection();
        try {
          PreparedStatement stmt = conn.prepareStatement(
              "select caller_id from " + store.getIndexQueueLockTableName() + " where directory_name=? and name=? order by arrival_time");
          try {
            stmt.setString(1, store.getDirectoryName());
            stmt.setString(2, name);
            ResultSet rs = stmt.executeQuery();
            try {
              if ( rs.next() ) {
                String currFirstCallerId = rs.getString( 1 );
                int cntr = 0;
                StringBuffer buff = new StringBuffer( cntr++ + "->" + currFirstCallerId );
                while ( rs.next() ) {
                  buff.append( ",").append(cntr++).append( "->").append( rs.getString( 1 ) );
                }

//System.out.println( Thread.currentThread().getName() + "::" + name + ")BlobDirectory::isFirstInQueue (" + buff + ")");
                if ( callerId.equals(currFirstCallerId) )
                  leave = true;
              } else
                leave = true;
            } finally {
              rs.close();
            }
          } finally {
            stmt.close();
          }
        } finally {
          store.closeConnection( conn );
        }
      } catch ( SQLException err ) {
        throw new IOException( err.getMessage() );
      }
      return leave;
    }

    private void leaveQueue() throws IOException {
//System.out.println( Thread.currentThread().getName() + "::" + name +")BlobDirectory::leaveQueue (" + callerId + ")");
      try {
        Connection conn = store.getConnection();
        try {
          PreparedStatement stmt = conn.prepareStatement(
              "delete from " + store.getIndexQueueLockTableName() + " where directory_name=? and name=? and caller_id=?");
//System.out.println( new java.util.Date() + "> " + callerId + " is trying to leave the queue [" + Thread.currentThread().getName() + "]:: " + name);
          try {
            stmt.setString(1, store.getDirectoryName());
            stmt.setString(2, name);
            stmt.setString(3, callerId);
            stmt.executeUpdate();
//System.out.println( new java.util.Date() + "> " + callerId + " has left the queue [" + Thread.currentThread().getName() + "]:: " + name);
          } finally {
            stmt.close();
          }
        } finally {
          store.closeConnection( conn );
        }
      } catch ( SQLException err ) {
//System.out.println( new java.util.Date() + "> " + callerId + " has had troubles leaving the queue [" + Thread.currentThread().getName() + "]:: " + name);
        throw new IOException( err.getMessage() );
      }
    }
  }

  private class BlobRowsInputStream extends InputStream  {
    private String name;
    int pointer = 0;

    public BlobRowsInputStream( String nm, long len ) {
      name = nm;
      length = len;
    }

    public void readInternal(byte[] dest, int destOffset, int len) throws IOException {
// System.out.println( Thread.currentThread().getName() + ")BlobDirectory.BlobRowsInputStream.readInternal, name=" + name + ", dOff=" + destOffset + ", len=" + len );
      int remainder = len;
      int start = pointer;
      try {
        Connection conn = store.getConnection();
        try {
          while (remainder != 0) {
            int bufferNumber = start/BUFFER_SIZE;
            int bufferOffset = start%BUFFER_SIZE;
            int bytesInBuffer = BUFFER_SIZE - bufferOffset;
            int bytesToCopy = bytesInBuffer >= remainder ? remainder : bytesInBuffer;
            byte[] buffer;
            buffer = getBuffer( conn, name, bufferNumber );

            System.arraycopy(buffer, bufferOffset, dest, destOffset, bytesToCopy);
            destOffset += bytesToCopy;
            start += bytesToCopy;
            remainder -= bytesToCopy;
          }
        } finally {
          store.closeConnection( conn );
        }
      } catch ( SQLException err ) {
        throw new IOException( err.getMessage() );
      }
      pointer += len;
    }

    public void close() {
//    	System.out.println( Thread.currentThread().getName() + ")BlobDirectory.BlobRowsInputStream.close name=" + name );
    }

    public void seekInternal(long pos) {
//    	System.out.println( Thread.currentThread().getName() + ")BlobDirectory.BlobRowsInputStream.seekInternal, name=" + name + ", pos=" + pos );
      pointer = (int)pos;
    }
  }

  private class TemporaryFileOutputStream extends OutputStream {
    private String name;
    private File tempFile;
    private RandomAccessFile file;

    public TemporaryFileOutputStream( String nm ) throws IOException {
      name = nm;
      tempFile = File.createTempFile( name, ".idx" );
// System.out.println("tempfile=" + tempFile );
      file = new RandomAccessFile( tempFile, "rw" );
    }

    public void flushBuffer(byte[] b, int size) throws IOException {
//System.out.print("(tmp.flushBuffer,th=" + Thread.currentThread().hashCode() + ")");
// System.out.println("<TemporaryFileOutputStream.flushBuffer>" );
// System.out.println("\tbuff=" + new String(b) );
// System.out.println("\tsize=" + size );
// System.out.println("</TemporaryFileOutputStream.flushBuffer>" );
      file.write(b, 0, size);
    }

    public void close() throws IOException {
// System.out.println("<TemporaryFileOutputStream.close>" );
      super.close();
      file.close();
      try {
        long length = tempFile.length();
        Connection conn = store.getConnection();
//System.out.print("(tmp.close,th=" + Thread.currentThread().hashCode() + "),cn=" + conn.hashCode() + ")" );
        try {
          if ( length > 0 )
            writeToBufferTable( conn );
          updateLength( conn, length );
        } finally {
          store.closeConnection( conn );
        }
      } catch (SQLException err ) {
        throw new IOException( err.getMessage() );
      } finally {
        tempFile.delete();
      }
// System.out.println("</TemporaryFileOutputStream.close>" );
    }

    private void updateLength( Connection conn, long newLen ) throws SQLException {
//System.out.print("(tmp.updateLength,th=" + Thread.currentThread().hashCode() + "),cn=" + conn.hashCode() + ")" );
      PreparedStatement stmt = conn.prepareStatement( "update " + store.getIndexTableName() + " set length=?, lastmodified=? where name=? and directory_name=?" );
      try {
        stmt.setLong(1, newLen);
        stmt.setTimestamp(2, new java.sql.Timestamp(System.currentTimeMillis()) );
        stmt.setString(3, name);
        stmt.setString(4, store.getDirectoryName());
        stmt.executeUpdate();
      } finally {
        stmt.close();
      }
    }

    public final void seek(long pos) throws IOException {
      super.seek(pos);
//System.out.print("(tmp.seek,th=" + Thread.currentThread().hashCode() + ")");
      file.seek(pos);
// System.out.println("<BlobRowsOutputStream.seek>" );
// System.out.println("\tpos=" + pos );
// System.out.println("</BlobRowsOutputStream.seek>" );
    }

    public final long length() throws IOException {
//System.out.print("(tmp.length,th=" + Thread.currentThread().hashCode() + ")");
// System.out.println("<TemporaryFileOutputStream.length>" );
// System.out.println("\tlen=" + file.length() );
// System.out.println("</TemporaryFileOutputStream.length>" );
      return file.length();
    }

//    protected void finalize() throws IOException {
//      file.close();
//    }
//
    private void writeToBufferTable( Connection conn ) throws IOException, SQLException {
//System.out.print("(tmp.writeToBufferTable,th=" + Thread.currentThread().hashCode() + "),cn=" + conn.hashCode() + ")" );
// System.out.println("TemporaryFileOutputStream.writeToBufferTable" );
      PreparedStatement stmt = conn.prepareStatement( "insert into " + store.getBufferTableName() + " (directory_name, name, id, buffer) values (?,?,?,?)" );
      try {
        FileInputStream fis = new FileInputStream( tempFile );
        BufferedInputStream bis = new BufferedInputStream( fis );
        try {
          byte[] transBuff = new byte[BUFFER_SIZE];
          int readIn;
          int id = 0;
          while ( (readIn = bis.read(transBuff)) > -1 ) {
            if ( readIn < BUFFER_SIZE )
              Arrays.fill( transBuff, Math.max(readIn,0), BUFFER_SIZE, (byte) 0 );
// System.out.println("TemporaryFileOutputStream.writeToBufferTable (nm=" + name + ",id=" + id + ", sz=" + readIn + ")" );
            stmt.setString(1, store.getDirectoryName());
            stmt.setString(2, name);
            stmt.setInt(3, id++);
            stmt.setBytes(4, transBuff);
            stmt.executeUpdate();
            stmt.clearParameters();
          }
        } finally {
          bis.close();
        }
      } finally {
        stmt.close();
      }
    }
  }
}